package cn.itcast.streaming

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Window {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[6]").setAppName("update state by key")
    val sc = new SparkContext(conf)
    sc.setLogLevel("ERROR")
    val ssc = new StreamingContext(sc, Seconds(1))

    val lines = ssc.socketTextStream(
      hostname = "192.168.88.100",
      port = 8888,
      storageLevel = StorageLevel.MEMORY_AND_DISK_SER
    )

    val words = lines.flatMap(_.split(" ")).map((_, 1))

    val wordCounts = words.reduceByKeyAndWindow(
      reduceFunc = (n, r) => n + r,
      windowDuration = Seconds(30),                 // 窗口长度，多长时间内的RDD合并为一个
      slideDuration = Seconds(10)                   // 滑动间隔，每个多久生成一个window
    )

    wordCounts.print()

    ssc.start()
    ssc.awaitTermination()

  }
}
